package jxkjsync

import (
	"context"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

type stepWorker struct {
	configSuffix           string
	mw                     WorkerWindow
	isRun                  bool
	isProcess              int32
	amBeginning            int32
	amEnding               int32
	amSaving               int32
	interval, times        int
	jobCtx                 context.Context
	jobCancel              context.CancelFunc
	endErr                 error
	jobList                []jobInfo
	jobLen                 int
	exceptionHandle        func(err error)
	beforeStart, afterStop func() error
	jobFinishHook          func() error
	logger                 Logger
	prefixSyncedCache      string
}

func (w *stepWorker) setLogger(logger Logger) {
	w.logger = logger
}
func (w *stepWorker) setPrefixSyncedCache(px string) {
	w.prefixSyncedCache = px
}
func (w *stepWorker) SetBeforeStart(f func() error)        { w.beforeStart = f }
func (w *stepWorker) SetAfterStop(f func() error)          { w.afterStop = f }
func (w *stepWorker) SetJobFinishHook(f func() error)      { w.jobFinishHook = f }
func (w *stepWorker) SetExceptionHandle(f func(err error)) { w.exceptionHandle = f }
func (w *stepWorker) HandleException(err error) {
	if w.exceptionHandle != nil {
		w.exceptionHandle(err)
	}
}
func (w *stepWorker) DoJobFinishHook() {
	if w.jobFinishHook != nil {
		w.jobFinishHook()
	}
}
func (w *stepWorker) DoAfterStop() {
	if w.afterStop != nil {
		w.afterStop()
	}
}
func (w *stepWorker) DoBeforeStart() error {
	if w.beforeStart != nil {
		return w.beforeStart()
	}
	return nil
}

func (w *stepWorker) IsRun() bool {
	return w.isRun
}
func (w *stepWorker) Error() error {
	return w.endErr
}
func (w *stepWorker) DoAwake(switchMap map[string]bool, sqlMap map[string]string) {
	if !w.isRun {
		w.DoStart(switchMap, sqlMap)
	}
}

func (w *stepWorker) DoStart(switchMap map[string]bool, sqlMap map[string]string) {
	go func() {
		defer fmt.Println("doStart end ")
		if !atomic.CompareAndSwapInt32(&w.amBeginning, 0, 1) {
			w.mw.NotifyInfo("同步开启中...")
			return
		}
		defer atomic.StoreInt32(&w.amBeginning, 0)
		isOk := false
		if w.isRun {
			w.mw.NotifyInfo("同步任务进行中")
			return
		}
		w.mw.ShowRunInfo("同步启动中..")
		w.mw.ShowReading()
		defer func() {
			if !isOk {
				//w.mw.ShowRunInfo("同步启动失败..")
				w.mw.ShowStopping()
			} else {
				w.isRun = true
				w.mw.ShowRunning()
			}
		}()
		w.mw.ShowRunInfo("同步任务初始化..")
		if err := w.DoBeforeStart(); err != nil {
			w.mw.NotifyInfo(err.Error())
			return
		}
		if len(switchMap) == 0 {
			w.mw.NotifyInfo("未开启同步项" + "(20009)")
			return
		}
		hasOpen := false
		for _, open := range switchMap {
			if open {
				hasOpen = open
			}
		}
		if !hasOpen {
			w.mw.NotifyInfo("未开启同步项" + "(20010)")
			return
		}

		//if lErr := initialize.LoadSqlConfig(); lErr != nil {
		//	w.logger.Error(lErr)
		//	w.mw.ShowRunInfo(lErr.Error())
		//	w.mw.NotifyInfo(lErr.Error())
		//	return
		//}

		jobList := make([]jobInfo, 0)
		jobLen := 0
		for _, t := range syncTypes {
			if open := switchMap[t]; open {
				srv := GetSyncService(t)
				if srv == nil {
					w.mw.NotifyInfo("暂不支持同步数据类型: " + t + "(20004)")
					return
				}
				if !srv.WantRun() {
					continue
				}

				jb := jobInfo{srv: srv}
				if srv.WantSql() {
					if _, ok := sqlMap[t]; !ok {
						w.mw.NotifyInfo("同步数据类型: " + srv.GetNameZh() + " 未配置查询语句" + "(20000)")
						return
					} else {
						if err := srv.ValidateSql(sqlMap[t]); err != nil {
							w.mw.NotifyInfo("同步数据类型: " + srv.GetNameZh() + " " + err.Error())
							return
						}
						jb.sql = sqlMap[t]
					}
				}
				//附属同步语句
				if srv.GetSubType() != "" {
					if GetSyncService(srv.GetSubType()).WantSql() {
						if _, ok := sqlMap[srv.GetSubType()]; !ok {
							w.mw.NotifyInfo("同步数据类型:" + (GetSyncService(srv.GetSubType())).GetNameZh() + " 未配置查询语句" + "(20000)")
							return
						}
						jb.sql = jb.sql + "&&" + sqlMap[srv.GetSubType()]
					}
				}
				jobList = append(jobList, jb)
				jobLen++
			}
		}
		if jobLen == 0 {
			w.mw.NotifyInfo("未开启同步项" + "(20013)")
			return
		}
		jobs := "will sync:"
		for _, step := range StepTypes {
			tt := ""
			num := 0
			for _, job := range jobList {
				if job.srv.GetStep() == step {
					tt += job.srv.GetNameZh() + "|"
					num++
				}
			}
			jobs += fmt.Sprintf("[odr:%d,len:%d,type:%s]", step, num, tt)
		}
		fmt.Println(jobs)
		w.logger.Info(jobs)
		fmt.Println(">>>>>>> len jobList:", jobLen)

		w.jobCtx, w.jobCancel = context.WithCancel(context.Background())
		w.jobList = jobList
		w.jobLen = jobLen
		go w.doSync()
		isOk = true
	}()
}

func (w *stepWorker) DoStop(err error) {
	go func() {
		defer fmt.Println("doStop end ")
		if !w.isRun {
			w.mw.NotifyInfo("同步未启动")
			return
		}
		if !atomic.CompareAndSwapInt32(&w.amEnding, 0, 1) {
			w.mw.NotifyInfo("同步正在结束中...")
			return
		}

		defer atomic.StoreInt32(&w.amEnding, 0)
		infoErr := "同步"
		if err != nil {
			infoErr += "异常"
		}
		w.mw.ShowRunInfo(infoErr + "停止中...")
		w.endErr = err
		ok := false
		w.mw.ShowReading()
		defer func() {
			if !ok {
				defer w.mw.ShowRunInfo(infoErr + "停止失败...")
				w.mw.ShowRunning()
			} else {
				defer w.mw.ShowRunInfo(infoErr + "已停止...")
				w.isRun = false
				w.mw.ShowStopping()
			}
		}()
		if w.jobCancel != nil {
			fmt.Println("stop call cancel")
			w.jobCancel()
		}
		//延迟更新结束状态 等待进行中任务完成
		for {
			if w.isProcess == 1 {
				fmt.Println("同步进程结束中-等候两秒")
				w.logger.Infof("同步进程结束中-请等候 %s", time.Now().Format("2006-01-02 15:04:05"))
				//w.mw.ShowRunInfo("同步任务结束中... ")
				time.Sleep(time.Second * 2)
			} else {
				break
			}
		}
		w.DoAfterStop()
		//initialize.CloseDb(g.DbErp)
		//initialize.CloseDb(g.DbShop)
		//cache.Close()
		ok = true
	}()
}

func (w *stepWorker) doSync() {
	defer fmt.Println("doSync end ")
	/////////////////////////新调度逻辑////////////////////
	errChx := make(chan error, w.jobLen*2)
	n := time.Now()
	delayTimesX := 0
	timerX := time.Tick(time.Duration(w.interval) * time.Second)
	w.times++
	jobIDPrefix := fmt.Sprintf("wk%s", fmt.Sprintf("%02d%02d%02d%02d%02d", n.Month(), n.Day(), n.Hour(), n.Minute(), n.Second()))
	jobID := fmt.Sprintf("%s%d-%d", jobIDPrefix, w.times, delayTimesX)
	w.doJob(jobID, errChx)
	for {
		select {
		case err := <-errChx:
			w.isRun = true
			w.DoStop(err)
			w.mw.ShowRunInfo(err.Error())
			w.mw.NotifyInfo(err.Error())
			return
		case <-w.jobCtx.Done():
			w.logger.Info("doSync done  received")
			fmt.Println("doSync done  received")
			return
		case <-timerX:
			if w.isProcess == 0 {
				w.times++
				jobID = fmt.Sprintf("%s%d-%d", jobIDPrefix, w.times, delayTimesX)
				go w.doJob(jobID, errChx)
			} else {
				delayTimesX++
			}
		}
	}

	return
	/////////////////////////新调度逻辑////////////////////
	//time.Sleep(time.Second * 2)
	errCh := make(chan error, w.jobLen*2)
	jobCh := make(chan string, 1)
	//defer close(errCh)
	defer close(jobCh)
	go w.loopJob(jobCh, errCh)
	//now := time.Now()
	// 年月日时分秒作为任务前缀
	jobPrefix := fmt.Sprintf("%04d%02d%02d%02d%02d%02d", n.Year(), n.Month(), n.Day(), n.Hour(), n.Minute(), n.Second())
	w.times = 1
	delayTimes := 0
	jobCh <- fmt.Sprintf("wk%s%d-%d", jobPrefix, w.times, delayTimes)
	timer := time.Tick(time.Duration(w.interval) * time.Second)

	for {
		select {
		case <-w.jobCtx.Done():
			fmt.Println("out   done  received")
			return
		case err := <-errCh:
			w.isRun = true
			w.DoStop(err)
			w.mw.ShowRunInfo(err.Error())
			w.mw.NotifyInfo(err.Error())
			return
		case <-timer:
			if len(jobCh) > 0 {
				delayTimes++
				fmt.Println("Current task in progress,continue")
			} else {
				w.times++
				jobCh <- fmt.Sprintf("wk%s%d-%d", jobPrefix, w.times, delayTimes)
			}
		}
	}
}
func (w *stepWorker) loopJob(jobQueue chan string, errCh chan error) {
	if !atomic.CompareAndSwapInt32(&w.isProcess, 0, 1) {
		w.logger.Info("unexcepted sync job2!!!")
		return
	}
	defer func() {
		atomic.StoreInt32(&w.isProcess, 0)
		fmt.Println("loopJob end ")
	}()
	//w.logger.Info("new job type len:", len(jobList))
	for {
		select {
		case jobID, ok := <-jobQueue:
			if !ok {
				w.logger.Info("job chan closed")
				fmt.Println("job chan closed")
				return
			}
			select {
			case <-w.jobCtx.Done():
				w.logger.Info("loopJob  out  done  received")
				fmt.Println("loopJob  out  done  received")
				return
			default:
				w.doJob(jobID, errCh)
			}
		}
	}
}
func (w *stepWorker) doJob(jobID string, errCh chan error) {
	//////////////////
	if !atomic.CompareAndSwapInt32(&w.isProcess, 0, 1) {
		w.logger.Info("unexpected sync job!!!")
		return
	}
	defer func() {
		atomic.StoreInt32(&w.isProcess, 0)
		//w.isProcess = false
		fmt.Println("doJob end ")
	}()
	/////////////////////////
	s := time.Now()
	errHappenCh := make(chan struct{}, w.jobLen*2)
	//isSuccess := 1

	defer func() {
		//if len(errHappenCh) > 0 {
		//	isSuccess = 0
		//}
		//_ = shop_query.SyncMetrics.Create(&shop_entity.SyncMetrics{
		//	JobID:       jobID,
		//	SyncService: "all-service",
		//	UsedSecond:  time.Since(s).Milliseconds(),
		//	StartTime:   s,
		//	EndTime:     time.Now(),
		//	IsSuccess:   int32(isSuccess),
		//	ErrorMsg:    "",
		//})
		close(errHappenCh)
	}()
	var (
		syncFailed int32 = 0
	)
	w.mw.ShowRunInfo("开始:" + jobID)
	for _, step := range StepTypes {
		select {
		case <-w.jobCtx.Done():
			w.logger.Info("syncData inner  done  received")
			fmt.Println("syncData inner  done  received")
			return
		default:
			wg := sync.WaitGroup{}
			for _, job := range w.jobList {
				if job.srv.GetStep() == step {
					wg.Add(1)
					go func(id, sql string, srv SyncService) {
						defer wg.Done()
						since := time.Now()
						defer func() {
							w.logger.Info(id, srv.GetNameZh(), " alone used second:", time.Since(since).Seconds())
						}()
						syncErrx := srv.SyncFunc(w.jobCtx, w.prefixSyncedCache+srv.GetName(), sql)
						//metric := &shop_entity.SyncMetrics{
						//	JobID:       id,
						//	SyncService: srv.GetName(),
						//	UsedSecond:  time.Since(since).Milliseconds(),
						//	StartTime:   since,
						//	EndTime:     time.Now(),
						//	IsSuccess:   1,
						//	ErrorMsg:    "",
						//}
						//if syncErr != nil {
						//	metric.IsSuccess = 0
						//	metric.ErrorMsg = syncErr.Error()
						//}
						//_ = shop_query.SyncMetrics.Create(metric)
						/////////////////
						if syncErrx != nil {
							atomic.StoreInt32(&syncFailed, 1)
							w.logger.Error()
							w.HandleException(fmt.Errorf("%s(%s) 信息: %s", srv.GetNameZh(), id, syncErrx.Error()))
							w.mw.ShowRunInfo(srv.GetNameZh() + id + " :异常!")
						}
						//action := "default callback:"
						//if syncErr != nil && strings.Contains(syncErr.Error(), "invalid connection") {
						//	w.logger.Error(action + srv.GetName() + id + " failed, error:" + syncErr.Error())
						//	action = "retry callback:"
						//	time.Sleep(time.Second * 2)
						//	syncErr = srv.SyncFunc(w.jobCtx, w.prefixSyncedCache+srv.GetName(), sql)
						//}
						//if syncErr != nil {
						//	w.logger.Error(action + srv.GetName() + id + " failed, error:" + syncErr.Error())
						//	errHappenCh <- struct{}{}
						//	errCh <- errors.New(srv.GetNameZh() + " 数据同步异常: " + syncErr.Error())
						//}
						//////////////////
					}(jobID, job.sql, job.srv)
				}
				//else {
				//	err := errors.New("暂不支持同步类型: " + job.dataType + "(20002)")
				//	fmt.Println(err)
				//	w.logger.Error(err)
				//	errCh <- err
				//	return
				//}
			}
			wg.Wait()
			if len(errHappenCh) > 0 {
				w.mw.ShowRunInfo(fmt.Sprintf("任务异常中断,step:%+v", step))
				return
			}
			w.mw.ShowRunInfo(fmt.Sprintf("⇄:%s(%+v)", jobID, step))
		}
		fmt.Println("syncData wait end order:", step)
		w.logger.Info(fmt.Sprintf("end job order:%d ,id:%s ", step, jobID))
	}
	time.Sleep(time.Second * 1)
	w.logger.Info(jobID, " all used second:", time.Since(s).Seconds())
	if syncFailed == 1 {
		w.mw.ShowRunInfo("失败:" + jobID)
		return
	}
	w.mw.ShowRunInfo("完成:" + jobID)
	{
		w.mw.ShowRunInfo("时间:" + time.Now().Format("06/01/02 15:04:05"))
		// 更新同步时间
		w.DoJobFinishHook()
		//_, _ = shop_query.Config.
		//	Where(shop_query.Config.ConfigKey.Eq("API_CONFIG")).
		//	Update(shop_query.Config.ModifyTime, time.Now().Unix())
	}
}
